package sanshui.system.trade.factory.trade;

import cn.hutool.core.bean.BeanUtil;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.logging.log4j.util.Strings;
import org.springframework.beans.BeanUtils;
import sanshui.system.trade.common.GlobalContain;
import sanshui.system.trade.common.TradeType;
import sanshui.system.trade.config.SpringContextHolder;
import sanshui.system.trade.db.entity.TradeWorker;
import sanshui.system.trade.factory.data.*;
import sanshui.system.trade.factory.trade.riskCtr.RiskCtrCode;
import sanshui.system.trade.factory.trade.strategy.StrategyCode;
import sanshui.system.trade.factory.trade.strategy.StrategyFactory;
import sanshui.system.trade.factory.trade.strategy.TradeStrategy;
import sanshui.system.trade.factory.trade.worker.*;
import sanshui.system.trade.factory.trade.worker.market.AGuBackTradeWorker;
import sanshui.system.trade.factory.trade.worker.market.AGuRtTradeWorker;
import sanshui.system.trade.model.*;
import sanshui.system.trade.service.IAccountService;
import sanshui.system.trade.service.IWorkerService;
import sanshui.system.trade.util.CommonTimeUtil;
import sanshui.system.trade.util.IdWorkerUtil;

import java.time.LocalDateTime;
import java.time.temporal.ChronoUnit;
import java.util.*;
import java.util.concurrent.*;

@Slf4j
public class TradeSystemExecutor {

    IWorkerService workerService = SpringContextHolder.getBean(IWorkerService.class);
    IAccountService accountService = SpringContextHolder.getBean(IAccountService.class);
    StrategyFactory strategyFactory = SpringContextHolder.getBean(StrategyFactory.class);

    Queue<Long> blockingDeque;

    ExecutorService threadPool;
    ExecutorCompletionService<WorkerResult> service;
    Map<Long, AbstractTradeWorker> workerMap = new HashMap<>();

    int numThread = 10;
    boolean stop = false;

    /**
     * 创建多线程回测，主要是策略在单支标的胜率
     */
    public void stopIt(){
        stop = true;
    }

    public void init(int numThread) {
        threadPool = Executors.newCachedThreadPool();
        service = new ExecutorCompletionService<>(threadPool);
        blockingDeque = new ArrayDeque<>(10);
        this.numThread = numThread;
    }

    public void loadWorker(){
        while (!stop){
            try {
                this.blockingDeque.addAll(workerService.selectWorkersIdByStatus(TradeType.WorkerStatus.INIT));
                Thread.sleep(10000L);
            }catch (Exception e){
                log.error("loader error", e);
            }
        }
    }

    public void run() {
        Long currentWorkId = null;
        while (!stop) {
            try {
                log.debug("当前正在运行的Worker数量:{}", workerMap.size());
                clearFailWorker();
                currentWorkId = this.blockingDeque.poll();

                // 检查数据库中进行中是否真实在运行，如果未真实在运行，更新其状态为结束
                // 如果是分布式到话就有问题了
                workerService.checkTaskIfRunInMap(workerMap);

                if (null == currentWorkId) {
                    Thread.sleep(5000L);
                    continue;
                }
                TradeWorker currentWorker = workerService.selectWorkerById(currentWorkId);
                log.info("当前worker:{}", JSONObject.toJSONString(currentWorker));
                int lock = workerService.updateWorkerStatusByWorkerId(currentWorkId, TradeType.WorkerStatus.IN, TradeType.WorkerStatus.INIT);
                if (lock == 0){
                    log.error("未拿到任务锁, workerId:{}", currentWorkId);
                    continue;
                }

                if (workerMap.size() >= this.numThread) {
                    log.info("大于并发数，需要等待，并发数量：{}", this.numThread);
                    Thread.sleep(10000L);
                    continue;
                }

                TradeWorker tradeWorker = workerService.selectWorkerById(currentWorkId);
                if (!TradeType.WorkerStatus.IN.getCode().equalsIgnoreCase(tradeWorker.getWorkerStatus())){
                    log.error("任务状态非进行中(=1), status:{}, workerId:{}", tradeWorker.getWorkerStatus(), currentWorkId);
                    continue;
                }

                TradeSystemType tradeSystemType = TradeSystemType.valueOf(tradeWorker.getTradeSystemType());
                TradeMarketKey tradeMarketKey = TradeMarketKey.valueOf(tradeWorker.getTradeMarketKey());

                AbstractTradeWorker tradeWorkerExe;

                if (!strategyFactory.isHas(StrategyCode.valueOf(tradeWorker.getStrategyCode()))){
                    log.info("未知的策略");
                    continue;
                }
                if (!strategyFactory.isHas(RiskCtrCode.valueOf(tradeWorker.getRiskCtr()))){
                    log.error("未知的风控策略");
                    continue;
                }

                TradeWorkerModel tradeWorkerModel = workerService.selectWorkerModelById(currentWorkId);
                log.info("准备运行worker");
                TradeSysWorkerModel modelInTrade = null;
                DataCache dataContainer = null;
                switch (tradeSystemType) {
                    case BACK_TRADE:
                        modelInTrade = this.buildRunTradeModelInBackTrade(tradeWorkerModel);
                        dataContainer = this.buildDataCache(tradeWorkerModel, modelInTrade);
                        switch (tradeMarketKey){
                            case A_GU:
                                tradeWorkerExe = new AGuBackTradeWorker(modelInTrade, dataContainer);
                                break;
                            case OKE:
                                log.error("交易【回测系统】, 不支持【OKE交易市场】类型");
                                continue;
                            default:
                                log.error("交易【回测系统】, 未知的【交易市场】类型");
                                continue;
                        }
                        break;
                    case RT:
                        switch (tradeMarketKey){
                            case A_GU:
                                modelInTrade = this.buildRunTradeModelInBackTrade(tradeWorkerModel);
                                DataCache dataCache = this.buildDataCacheRt(tradeWorkerModel, modelInTrade);
                                tradeWorkerExe = new AGuRtTradeWorker(modelInTrade, dataCache);
                                break;
                            case OKE:
                                log.error("交易【实时系统】, 不支持【OKE交易市场】类型");
                                continue;
                            default:
                                log.error("交易【实时系统】, 未知的【交易市场】类型");
                                continue;
                        }
                        break;
                        default:
                            log.info("未知的交易系统类型");
                            continue;
                }
                service.submit(tradeWorkerExe);
                workerMap.put(modelInTrade.getWorkerId(), tradeWorkerExe);
            } catch (Exception e) {
                log.error("发生错误, 自动弹出tradeModel", e);
                if (null!=currentWorkId){
                    workerService.updateWorkerStatusByWorkerId(currentWorkId, TradeType.WorkerStatus.END, TradeType.WorkerStatus.INIT);
                    workerService.updateWorkerStatusByWorkerId(currentWorkId, TradeType.WorkerStatus.END, TradeType.WorkerStatus.IN);
                }
            } finally {
            }

        }
    }

    private TradeAccountModel buildTradeAccountModel(TradeWorkerModel tradeWorkerModel){
        TradeAccountModel accountModel = new TradeAccountModel();
        BeanUtil.copyProperties(tradeWorkerModel, accountModel);
        return accountModel;
    }

    private void clearFailWorker(){
        for (Map.Entry<Long, AbstractTradeWorker> entry : workerMap.entrySet()) {
            Long workerId = entry.getKey();
            AbstractTradeWorker worker = entry.getValue();
            if (!worker.isRunning()){
                workerMap.remove(workerId);
                workerService.updateWorkerStatusByWorkerId(workerId, TradeType.WorkerStatus.END, TradeType.WorkerStatus.INIT);
                log.info("Worker:{}, acc:{},运行完成, 从List中移除", worker.getWorkerName(), worker.getAccId());
                worker = null;
            } else {
                TradeWorker tradeWorker = workerService.selectWorkerById(workerId);
                String startTime = tradeWorker.getUpdateTime();
                String workerStatus = tradeWorker.getWorkerStatus();
                LocalDateTime start = CommonTimeUtil.getTimestamp(startTime);
                boolean isTimeout = CommonTimeUtil.compare(start, LocalDateTime.now(), 30*60, ChronoUnit.SECONDS);
                if (isTimeout && !TradeType.WorkerStatus.END.getCode().equalsIgnoreCase(workerStatus)
                && TradeSystemType.BACK_TRADE.name().equalsIgnoreCase(tradeWorker.getTradeSystemType())){
                    log.error("任务超时: workerId:{}, accId:{}", workerId, tradeWorker.getAccId());
                    worker.stop();
                }

            }

        }
    }

    /**
     * 回测交易系统
     * */
    private TradeSysWorkerModel buildRunTradeModelInBackTrade(TradeWorkerModel tradeWorkerModel){
        return TradeSysWorkerModel.builder()
                .workerId(tradeWorkerModel.getWorkerId())
                .startTime16(CommonTimeUtil.getNowTime16())
                .tradeAccountModel(buildTradeAccountModel(tradeWorkerModel))
                .tradeSystemType(TradeSystemType.valueOf(tradeWorkerModel.getTradeSystemType()))
                .tradeMarketKey(TradeMarketKey.valueOf(tradeWorkerModel.getTradeMarketKey()))
                .dataMarketType(DataMarketType.valueOf(tradeWorkerModel.getDataMarketType()))
                .workerName(Strings.isEmpty(tradeWorkerModel.getTradeName())? "DEFAULT_WORKER_".concat(CommonTimeUtil.getNowTime16()):tradeWorkerModel.getTradeName())
                .targetCodes(tradeWorkerModel.getChooseTargets())
                .targetCodesIndex(tradeWorkerModel.getChooseTargetsIndex())
                .indicatorCodes(tradeWorkerModel.getIndicators())
                .tradeStrategyCode(StrategyCode.valueOf(tradeWorkerModel.getStrategy()))
                .riskCtrCode(RiskCtrCode.valueOf(tradeWorkerModel.getRiskStrategy()))
                .frequency(tradeWorkerModel.getFrequency())
                .millis(tradeWorkerModel.getMillis() == null?0L:tradeWorkerModel.getMillis()).build();
    }

    private DataCache buildDataCache(TradeWorkerModel tradeWorkerModel, TradeSysWorkerModel buildTradeModel) throws Exception {
        DataLoadModel defaultDataLoadModel = new DataLoadModel();
        defaultDataLoadModel.setDataMarketType(buildTradeModel.getDataMarketType());
        defaultDataLoadModel.setTradeMarket(buildTradeModel.getTradeMarketKey());
        defaultDataLoadModel.setTradeSystemType(buildTradeModel.getTradeSystemType());
        defaultDataLoadModel.setTargetCodes(tradeWorkerModel.getChooseTargets());
        defaultDataLoadModel.setStartTime(CommonTimeUtil.getTimestamp(tradeWorkerModel.getStartTime()));
        defaultDataLoadModel.setEndTime(Strings.isEmpty(tradeWorkerModel.getEndTime())? null : CommonTimeUtil.getTimestamp(tradeWorkerModel.getEndTime()));
        defaultDataLoadModel.setFrequency(tradeWorkerModel.getFrequency());
        return new DataCache(defaultDataLoadModel);
    }

    private DataCache buildDataCacheRt(TradeWorkerModel tradeWorkerModel, TradeSysWorkerModel buildTradeModel) throws Exception {
        DataLoadModel defaultDataLoadModel = new DataLoadModel();
        defaultDataLoadModel.setDataMarketType(buildTradeModel.getDataMarketType());
        defaultDataLoadModel.setTradeMarket(buildTradeModel.getTradeMarketKey());
        defaultDataLoadModel.setTradeSystemType(buildTradeModel.getTradeSystemType());
        defaultDataLoadModel.setStartTime(CommonTimeUtil.getTimestamp(tradeWorkerModel.getStartTime()));
        defaultDataLoadModel.setEndTime(Strings.isEmpty(tradeWorkerModel.getEndTime())? null : CommonTimeUtil.getTimestamp(tradeWorkerModel.getEndTime()));
        defaultDataLoadModel.setFrequency(tradeWorkerModel.getFrequency());
        defaultDataLoadModel.setUserId(tradeWorkerModel.getUserId());
        defaultDataLoadModel.setWorkerId(tradeWorkerModel.getWorkerId());
        defaultDataLoadModel.setAccId(tradeWorkerModel.getAccId());
        return new DataCache(defaultDataLoadModel);
    }

    public AbstractTradeWorker getWorkerByWorkerId(String workerId){
        return this.workerMap.get(workerId);
    }
}
